//
// Created by tongj on 2023/10/7.
//

#include <MessageQueue/MessageQueue.h>
#include <base/log.hpp>
#include <base/time_utils.hpp>

#include <thread>

using namespace std;
using namespace chrono;
using namespace time_utils;
using namespace MessageQueue;
using namespace tbs;

using guard = std::unique_lock<std::mutex>;

timeUnit getTimeOut(CONST DelayMessage &c) {
  return c.getTime() - utils_now();
}

void MsgQueue::send(MessageQueue::DelayMessage &&item) {
  ASSERT((item._owner != nullptr && item._owner != this), "this msg has owner");
  {
	guard g(_w_mutex);
	item._owner = this;
	_q.emplace(std::move(item));
  }
  _cv.notify_one();
}

bool MsgQueue::isEmpty() CONST {
  return _q.empty();
}

void MsgQueue::cancel() {
  _cancel_flag = true;
  _cv.notify_one();
}

void MsgQueue::execute() {

  _run_flag = true;
  _cancel_flag = false;
  timeUnit span;
  while (!_cancel_flag) {
	// 空等待
	while (_q.empty() && !_cancel_flag) {
	  std::this_thread::yield();
	}
	if (_cancel_flag) {
	  break;
	}

	DelayMessage *dm = nullptr;

	// 时间等待
	{
	  guard w(_w_mutex);
	  dm = new DelayMessage(std::move(const_cast<DelayMessage &>(_q.top())));
	  _q.pop();
	}
	span = getTimeOut(*dm);
	// 时间等待
	if (span > 0) {
	  guard r(_r_mutex);
	  if (_cv.wait_for(r, standard_time(span)) != std::cv_status::timeout) {
		this->send(std::move(*dm));
		continue;
	  }
	}
	dm->consume();
	delete dm;
	dm = nullptr;
  }
  _run_flag = false;
}

bool MsgQueue::is_execute() CONST {
  return _run_flag;
}
size_t MsgQueue::messageCount() const {
  return _q.size();
}
